package com.shiwaixiangcun.monitor.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.shiwaixiangcun.core.plugin.dto.HashMonkeyDto;
import com.shiwaixiangcun.core.plugin.dto.MonkeyDto;
import com.shiwaixiangcun.monitor.enums.HealthStatus;
import com.shiwaixiangcun.monitor.enums.WarningType;
import com.shiwaixiangcun.monitor.service.EarlyWarningService;
import com.shiwaixiangcun.monitor.utils.MsConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetSocketAddress;
import java.util.List;

//@Component
public class CanalDataSynchronizer {

    private Logger logger = LoggerFactory.getLogger(CanalDataSynchronizer.class);

    @Value("${canal.host}")
    private String host;

    @Value("${canal.port}")
    private Integer port;

    @Value("${canal.username}")
    private String username = "";

    @Value("${canal.password}")
    private String password = "";

    @Value("${canal.destination}")
    private String destination = "example";

    @Value("${canal.schema}")
    private String schema = "sm_ly_main";

    @Value("${canal.table}")
    private String table = "sm_main";

    @Value("${canal.thread-sleep}")
    private Integer threadSleep = 1000;

    @Autowired
    private EarlyWarningService earlyWarningService;

    private Boolean running = true;

    @PreDestroy
    public void destroy() {
        this.running = false;
    }

    public void startSync() {
        // 创建链接
        InetSocketAddress address = new InetSocketAddress(host, port);
        CanalConnector connector = CanalConnectors.newSingleConnector(address, destination, username, password);
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (running) {
                try {
                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        try {
                            Thread.sleep(threadSleep);
                        } catch (InterruptedException ignored) {
                        }
                    } else {
                        // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                        printEntry(message.getEntries());
                    }
                    connector.ack(batchId); // 提交确认
                    // connector.rollback(batchId); // 处理失败, 回滚数据
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } finally {
            connector.disconnect();
        }
    }


    private void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            String schemaName = entry.getHeader().getSchemaName();
            String tableName = entry.getHeader().getTableName();
            if (!schema.contains(schemaName) || !table.contains(tableName)) {
                continue;
            }
            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }
            CanalEntry.EventType eventType = rowChage.getEventType();
            logger.info(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    schemaName, tableName,
                    eventType));

            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
//                if (eventType == CanalEntry.EventType.DELETE) {
//                    printColumn(rowData.getBeforeColumnsList());
//                } else if (eventType == CanalEntry.EventType.INSERT) {
//                    printColumn(rowData.getAfterColumnsList());
//                } else {
//                    logger.info("-------&gt; before");
//                    printColumn(rowData.getBeforeColumnsList());
//                    logger.info("-------&gt; after");
//                    printColumn(rowData.getAfterColumnsList());
//                }
                if (eventType == CanalEntry.EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private void printColumn(List<CanalEntry.Column> columns) {
        MonkeyDto monkeyDto = convertToMonkeyDto(columns);
        if ("06".equals(monkeyDto.getString("Keys_ID"))) {
            earlyWarningService.addEarlyWarning(monkeyDto);
        }
    }

    private MonkeyDto convertToMonkeyDto(List<CanalEntry.Column> columns) {
        MonkeyDto monkeyDto = new HashMonkeyDto();
        for (CanalEntry.Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            if ("Hardware_ID".equals(column.getName())) {
                monkeyDto.put("identifier", column.getValue());
            } else if ("Hardware_Name".equals(column.getName())) {
                monkeyDto.put("name", column.getValue());
            } else if ("Keys_ID".equals(column.getName())) {
                monkeyDto.put("Keys_ID", column.getValue());
            }
        }
        monkeyDto.put("waringType", WarningType.Calling_Device.name());
        monkeyDto.put("status", HealthStatus.DANGER);
        monkeyDto.put("content", "呼叫器呼叫");
        monkeyDto.put("orgId", MsConstants.DEFAULT_ORG_ID);
        return monkeyDto;
    }


}
